package com.lagou.client.boot;

import com.lagou.client.RPCConsumer;
import com.lagou.service.IUserServer;
import com.lagou.util.ZkUtil;
import org.I0Itec.zkclient.ZkClient;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class ConsumerBoot {

    private static Map<String, IUserServer> map = new HashMap<>();
    private static List<String> serverIpList = new ArrayList<>();

    private static ZkClient zkClient = new ZkClient(ZkUtil.ZK_ADDRESS);

    public static void main(String[] args) throws Exception {

        // 监听zk watcher通知
        subZk();

        serverIpList = getServerIpList();
        for (String address : serverIpList) {
            IUserServer server = (IUserServer) RPCConsumer.createProxy(IUserServer.class, address);
            map.put(address, server);
        }

        while (true) {
            try {
                IUserServer server = map.get(serverIpList.get(0));
                String s = server.sayHello("send client heartbeat......");
                Thread.sleep((long) (Math.random() * 5000));
            } catch (Exception e) {
                if (serverIpList.size() == 0) {
                    System.out.println("当前服务列表不存在，重连中");
                } else {
                    System.out.println("服务端 " + serverIpList.get(0) + "连接失败，重连中");
                }
                Thread.sleep(5000L);
            }
        }
    }

    /**
     * 获取服务器地址列表
     *
     * @return
     */
    private static List<String> getServerIpList() {
        List<String> children = null;
        if (zkClient.exists(ZkUtil.PATH)) {
            children = zkClient.getChildren(ZkUtil.PATH);
        }
        return children;
    }

    public static void subZk() {
        zkClient.subscribeChildChanges(ZkUtil.PATH, (parentPath, currentChildren) -> {
            serverIpList = currentChildren;

            for (String child : currentChildren) {
                // 说明有服务端上线
                if (map.getOrDefault(child, null) == null) {
                    System.out.println("zk watcher====>" + child + "：服务新上线");
                    IUserServer server = (IUserServer) RPCConsumer.createProxy(IUserServer.class, child);
                    map.put(child, server);
                }
            }

            for (Map.Entry<String, IUserServer> entry : map.entrySet()) {
                if (!currentChildren.contains(entry.getKey())) {
                    map.remove(entry.getKey());
                    System.out.println("zk watcher====>" + entry.getKey() + "：服务断开");
                }
            }
        });
    }
}
